package com.fyy.kafuka.utils;

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

/**
 * @Author fyy
 * @Description
 * @Date 2020/10/9 14:36
 */
@Component
public class KafkaConsumer {

    private static Logger logger = LoggerFactory.getLogger(KafkaConsumer.class);

    @KafkaListener(topics = "aaa")
    public void ConsumerMsg(ConsumerRecord<String,Object> record, @Header(KafkaHeaders.RECEIVED_TOPIC) String topic
            , Consumer consumer, Acknowledgment ack){
        logger.info("消费消息:topic={},context={}",topic,record.value());
        ack.acknowledge();
    }

}
